package demo;

import lombok.extern.slf4j.Slf4j;
import org.rocksdb.*;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.nio.ByteBuffer;
import java.time.LocalDate;
import java.util.Map;
import java.util.Scanner;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.function.Supplier;

@Slf4j
@SpringBootApplication
public class KafkaDemoApplication {



    public static class Reply {

    }



    public static void main(String[] args) {




        new Scanner(System.in).nextLine();





        SpringApplication.run(KafkaDemoApplication.class, args);

        int mils = 10023;

        KafkaDelayedQueue queue = new KafkaDelayedQueue(KafkaDelayedQueue.MqDelayedConfig.builder()
                .topic(String.format("hello%dx", mils))
                .delayMs(mils)
                .bootstrapServers("10.94.44.115:9092")
                .build());

        try {
            queue.start(String.format("hello%dx", mils), 4, 3000, (k, v) -> log.info("hello: key: {}, value: {}", k, v));
        } catch (KafkaDelayedQueue.DelayedQueueException e) {
            e.printStackTrace();
        }

        for(int i = 1;i < 5; ++i) {
            final AtomicInteger count = new AtomicInteger(i * 10000);

            new Thread(() -> {
                while (true) {
                    try {
                        queue.offer(count.incrementAndGet() + "-" + UUID.randomUUID().toString(), UUID.randomUUID().toString());
                    } catch (KafkaDelayedQueue.DelayedQueueException e) {
                        e.printStackTrace();
                    }
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }
}
